package com.shujia.core

import org.apache.flink.api.common.state.{ReducingStateDescriptor, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.util.Collector
import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.api.common.state.ReducingState

object Demo7Process {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val ds = env.socketTextStream("node1", 8888)

    ds.map(w => (w, 1))
      .keyBy(_._1)
      .process(new MyKeyedProcessFunction) //相当于flatmap
      .print()

    env.execute()
  }
}

class MyKeyedProcessFunction extends KeyedProcessFunction[String, (String, Int), (String, Int)] {

  var countState: ReducingState[Int] = _

  override def open(parameters: Configuration): Unit = {
    val reduce = new ReduceFunction[Int] {
      override def reduce(value1: Int, value2: Int): Int = {
        value1 + value2
      }
    }

    //用于统计数量的状态
    val countDesc = new ReducingStateDescriptor[Int]("count", reduce, classOf[Int])
    countState = getRuntimeContext.getReducingState(countDesc)
  }

  override def processElement(value: (String, Int), ctx: KeyedProcessFunction[String, (String, Int), (String, Int)]#Context, out: Collector[(String, Int)]): Unit = {

    //累加一次
    countState.add(1)
    if (countState.get() == 10) {
      //启动定时器   当达到定时器的时间之后触发onTimer 执行
      ctx.timerService().registerProcessingTimeTimer(System.currentTimeMillis())
    }
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, (String, Int), (String, Int)]#OnTimerContext, out: Collector[(String, Int)]): Unit = {

    val count = countState.get()

    //清空状态
    countState.clear()

    out.collect((ctx.getCurrentKey, count))
  }
}
